%%%-------------------------------------------------------------------
%%% File    : vect_h3.erl
%%% Author  :  <vjache>
%%% Description : 
%%%
%%% Created : 28 Jun 2009 by  <vjache>
%%%-------------------------------------------------------------------
-module(vect_h3).

-include("../include/util.hrl").
-include("../include/vect.hrl").
-include("../include/pipe.hrl").
-include("../include/iter.hrl").
-include("../include/htree.hrl").

-include_lib("eunit/include/eunit.hrl").

%% API
-export([
		 new/3,
		 open/3,
		 open_pipe/1,
		 close/1,
		 delete/2,
		 insert/2,
		 lookup/2,
		 delete_key/2,
		 all/1,
		 clear/1,
		 print_log/1]).

-record(vh3_state,{h3conf,name,arity}).
%%====================================================================
%% API
%%====================================================================

%%--------------------------------------------------------------------
%% Function: 
%% Description:
%%--------------------------------------------------------------------
new(H3Conf, Arity, Name) when is_record(H3Conf, h3conf),
							  is_integer(Arity),
							  is_atom(Name) ->
	#vect{
		 drv_module=?MODULE,
		 open_fun={open,[H3Conf, Arity, Name]},
		 open_pipe_fun=open_pipe,
		 close_fun=close,
		 delete_fun=delete,
		 delete_key_fun=delete_key,
		 insert_fun=insert,
		 lookup_fun=lookup,
		 all_fun=all,
		 clear_fun=clear
		 }.

open(H3Conf, Arity, Name) ->
	case H3Conf#h3conf.root_dir of
		{M,F,A} ->
			#vh3_state{
			   h3conf=H3Conf#h3conf{root_dir=apply(M,F,A)},
			   name=Name,
			   arity=Arity};
		_ ->
			#vh3_state{
			   h3conf=H3Conf,
			   name=Name,
			   arity=Arity}
	end.

open_pipe(Sess) when is_record(Sess,vh3_state) ->
	pipe:open(get_pipe(Sess)).

get_pipe(Sess) ->
	H3Conf=Sess#vh3_state.h3conf,
	Arity=Sess#vh3_state.arity,
	Tag=Sess#vh3_state.name,
	Pipe=htree:pipe_uploader(
		   H3Conf, 
		   _KeyFun=fun(Tuple)->
						   element(2,Tuple)
				   end),
	pipe_filter:new(
	  fun(Record) when 
		   is_tuple(Record),
		   tuple_size(Record)==(Arity+1),
		   element(1,Record)==Tag ->
			  true;
		 (BadRecord) ->
			  throw({bad_record,BadRecord,{sess,Sess}})
	  end, Pipe).

close(Sess) when is_record(Sess,vh3_state) ->
	ok.

lookup(Sess, Key) when is_record(Sess,vh3_state) ->
	H3Conf=Sess#vh3_state.h3conf,
	Tag=Sess#vh3_state.name,
	htree:lookup_by(
	  H3Conf, 
	  Key, 
	  _PredFun=fun(Record)->
					   element(1,Record)==Tag andalso element(2,Record)==Key
			   end).
insert(Sess, Tuple) when 
  is_record(Sess,vh3_state) andalso 
  is_tuple(Tuple) andalso 
  element(1,Sess#vh3_state.name) andalso 
  tuple_size(Tuple) == Sess#vh3_state.arity+1 ->
%% 	io:format("INSERT TUPLE~n"),
	insert(Sess, [Tuple]);
insert(Sess, Tuples) when 
  is_record(Sess,vh3_state),
  is_list(Tuples) ->
%% 	io:format("INSERT TUPLES~n"),
	insert(Sess, iter:from_list(Tuples));
insert(Sess, TupleIter) when 
  is_record(Sess,vh3_state),
  is_record(TupleIter,iter) ->
	Pump=pump:new(TupleIter, get_pipe(Sess)),
	ok=pump:init(Pump).
delete_key(Sess, Key) when is_record(Sess,vh3_state) ->
	H3Conf=Sess#vh3_state.h3conf,
	Tag=Sess#vh3_state.name,
	htree:delete_by(
	  H3Conf, 
	  Key, 
	  _PredFun=fun(Record)->
					   element(1,Record)==Tag andalso element(2,Record)==Key
			   end),
	ok.
delete(Sess, Tuple) when 
  is_record(Sess,vh3_state),
  is_tuple(Tuple) ->
	delete(Sess, [Tuple]);
delete(Sess, Tuples) when 
  is_record(Sess,vh3_state),
  is_list(Tuples) ->
	Tag=Sess#vh3_state.name,
	Dict0=lists:foldl(
			fun(Record,AccIn) ->
					case element(1,Record) of
						Tag ->
							Key=element(2,Record),
							dict:append(Key, Record, AccIn);
						_ ->
							AccIn
					end
			end, dict:new(), Tuples),
	Dict1=dict:map(
			fun(_Key,Records) ->
					RecordSet=sets:from_list(Records),
					_PredFun=fun(Record) ->
									 sets:is_element(Record, RecordSet)
							 end
			end, Dict0),
	H3Conf=Sess#vh3_state.h3conf,
	dict:fold(
	  fun(Key, PredFun, _AccIn) ->
			  htree:delete_by(H3Conf, Key, PredFun)
	  end, [], Dict1),
	ok.

all(Sess) when is_record(Sess,vh3_state) ->
	H3Conf=Sess#vh3_state.h3conf,
	Tag=Sess#vh3_state.name,
	iter_filter:new(
	  htree:all_iter(H3Conf), 
	  fun(Record) ->
			  element(1,Record)==Tag
	  end).

clear(Sess) when is_record(Sess,vh3_state) ->
	H3Conf=Sess#vh3_state.h3conf,
	ok=htree:truncate(H3Conf).
%%====================================================================
%% Internal functions
%%====================================================================

%%====================================================================
%% Test functions
%%====================================================================

basic_test_() ->
	{foreach, 
	 local, 
	 fun setup_/0, 
	 fun teardown_/1, 
	 [fun(VectSess)->
			  [
			   ?_test(test_insert(VectSess)),
			  ?_test(test_all(VectSess)),
			  ?_test(test_lookup(VectSess)),
			   ?_test(test_delete_key(VectSess)),
			   ?_test(test_delete(VectSess)),
			   ?_test(test_clear(VectSess))]
	  end]}.

test_insert(VectSess) ->
	vect:insert(VectSess, data()),
	timer:sleep(1000).
test_all(VectSess) ->
	?assertEqual(lists:sort(data()),
	lists:sort(iter:to_list(vect:all(VectSess)))).
test_lookup(VectSess) ->
	S0=lists:sort(select(s1)),
	S1=lists:sort(iter:to_list(vect:lookup(VectSess,s1))),
	?assertEqual(S0,S1).
test_delete_key(VectSess) ->
	vect:delete_key(VectSess,s1),
	S1=lists:sort(iter:to_list(vect:lookup(VectSess,s1))),
	?assertEqual([],S1).
test_delete(VectSess) ->
	Q1={spog,ss,p3,o9,test},
	Q2={spog,s3,p2,o6,test},
	vect:delete(VectSess,[Q1,Q2]),
	L1=iter:to_list(vect:lookup(VectSess,ss)),
	?assertEqual([],lists:filter(
	  fun(Q) ->
			  Q==Q1
	  end, L1)),
	L2=iter:to_list(vect:lookup(VectSess,s3)),
	?assertEqual([],lists:filter(
	  fun(Q) ->
			  Q==Q2
	  end, L2)).

test_clear(VectSess) ->
	vect:clear(VectSess),
	timer:sleep(2000),
	?assertEqual(0,
	length(iter:to_list(vect:all(VectSess)))).

setup_() ->
	timer:sleep(3010),
	RootDir=lists:flatten(io_lib:format("./test~p",[erlang:phash2(now())])),
%%  	RootDir=lists:flatten("./test1"),
	FileName="bck.log",
	QueueName=h3,
	DiskLogPoolName=disk_log_pool,
	H3Conf=htree:config_std(RootDir, FileName, QueueName, DiskLogPoolName),
	vect:open(new(H3Conf, 4, spog)).
teardown_(_VectSess) ->
	timer:sleep(100).

select(S) ->
	lists:filter(
	  fun({_,S1,_,_,_}) ->
			  S==S1
	  end, data()).
data() ->
	[
	 {spog,s1,p1,o1,test},
	 {spog,s1,p2,o2,test},
	 {spog,s1,p3,o3,test},
	 {spog,s2,p1,o1,test},
	 {spog,s2,p2,o4,test},
	 {spog,s3,p1,o5,test},
	 {spog,s3,p2,o6,test},
	 {spog,ss,p1,o7,test},
	 {spog,ss,p2,o8,test},
	 {spog,ss,p3,o9,test}
	 ].

print_log(Filename) ->
	iter:print(iter_disk_log:new([{name,print},{file,Filename}])).